package cn.xuexiyuan.flinkstudy.source;

import cn.xuexiyuan.flinkstudy.entity.Student;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;


/**
 * @Description:  自定义数据源 —— Mysql
 *
 * @Author 左龙龙
 * @Date 21-3-23
 * @Version 1.0
 **/
public class SourceDemo05_Costomer_Mysql {
    public static void main(String[] args) throws Exception {
        // 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 1.source
        DataStreamSource<Student> mysqlDS = env.addSource(new MySqlSource())
                .setParallelism(1);

        // 2.transformation

        // 3.sink
        mysqlDS.print();

        // 4.excute
        env.execute("SourceDemo05_Costomer_Mysql");
    }
}

class MySqlSource extends RichParallelSourceFunction<Student> {
    private boolean flag = true;
    private Connection connection = null;
    private PreparedStatement ps = null;
    private ResultSet rs = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        String url = "jdbc:mysql://localhost:3306/flink_study";
        String user = "root";
        String passwd = "";

        Class.forName("com.mysql.jdbc.Driver");
        connection = DriverManager.getConnection(url, user, passwd);
        ps = connection.prepareStatement("select id, name, age from tb_student");
    }

    @Override
    public void run(SourceContext<Student> sourceContext) throws Exception {

        while (flag){
            rs = ps.executeQuery();
            while (rs.next()){
                int id = rs.getInt("id");
                String name = rs.getString("name");
                int age = rs.getInt("age");

                sourceContext.collect(new Student(id, name, age));
            }

            Thread.sleep(5 * 1000);
        }
    }

    // 执行 cancel 的时候执行
    @Override
    public void cancel() {
        flag = false;
    }


    @Override
    public void close() throws Exception {
        if(null != connection) connection.close();
        if(null != ps) ps.close();
        if(null != rs) rs.close();
    }
}

